package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * Author: Felix
 * Date: 2021/11/26
 * Desc: 自定义反序列化器
 */
public class MyCustomSchema implements DebeziumDeserializationSchema<String> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> out) throws Exception {
        Struct valueStruct = (Struct) sourceRecord.value();
        Struct sourceStruct = valueStruct.getStruct("source");

        //获取数据库名
        String database = sourceStruct.getString("db");
        //获取表名
        String table = sourceStruct.getString("table");
        //操作类型的处理
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if ("create".equals(type)) {
            type = "insert";
        }

        //影响的数据行的处理
        Struct afterStruct = valueStruct.getStruct("after");
        //创建一个json对象，用于接收影响数据行
        JSONObject dataJsonObj = new JSONObject();
        if (afterStruct != null) {
            //获取afterstruct结构体中有哪些属性
            List<Field> fieldList = afterStruct.schema().fields();
            //对结构体中所有属性进行遍历
            for (Field field : fieldList) {
                //获取属性名称
                String fieldName = field.name();
                //获取属性的值
                Object fieldValue = afterStruct.get(field);
                //将遍历出来的属性名和值  放到json对象中
                dataJsonObj.put(fieldName, fieldValue);
            }
        }

        //整体返回的json对象的封装
        JSONObject resJsonObj = new JSONObject();
        resJsonObj.put("database", database);
        resJsonObj.put("table", table);
        resJsonObj.put("type", type);
        resJsonObj.put("data", dataJsonObj);

        out.collect(resJsonObj.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(String.class);
    }
}